package com.atguigu.gmall.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.bean.TrafficPageViewBean;
import com.atguigu.gmall.realtime.util.ClickhouseUtil;
import com.atguigu.gmall.realtime.util.DateFormatUtil;
import com.atguigu.gmall.realtime.util.MyKafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.Date;

public class DwsTrafficVcChArIsNewPageViewWindow {

    //1 、获取3条流
    //
    //2、把3条流的结构统一，统一成宽表结构
    //
    //
    //3 通过union把三条流合成一个宽表流
    //
    // 4  宽表的聚合
    public static void main(String[] args) throws Exception {

        // 0 环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);
        //1 、获取3条流

        String pageTopic = "dwd_traffic_page_log";
        String uvTopic = "dwd_traffic_uv_log";
        String ujTopic = "dwd_traffic_user_jump_out";
        String groupId = "dwd_traffic_page_window";
        //1.1  page流
        DataStreamSource<String> pageKafkaStream = env.addSource(MyKafkaUtil.getKafkaConsumer(pageTopic, groupId));
        DataStreamSource<String> uvKafkaStream = env.addSource(MyKafkaUtil.getKafkaConsumer(uvTopic, groupId));
        DataStreamSource<String> ujKafkaStream = env.addSource(MyKafkaUtil.getKafkaConsumer(ujTopic, groupId));

        //2、把3条流的结构统一，统一成宽表结构
        //2.1 page流的转换
        SingleOutputStreamOperator<TrafficPageViewBean> pageTrafficStream = pageKafkaStream.map(new MapFunction<String, TrafficPageViewBean>() {
            @Override
            public TrafficPageViewBean map(String jsonStr) throws Exception {
                JSONObject jsonObject = JSON.parseObject(jsonStr);
                JSONObject commonObj = jsonObject.getJSONObject("common");
                JSONObject pageObj = jsonObject.getJSONObject("page");
                Long ts = jsonObject.getLong("ts");
                Long sv = 0L;
                if (pageObj.getString("last_page_id") == null || pageObj.getString("last_page_id").length() == 0) {
                    sv = 1L;
                }

                TrafficPageViewBean trafficPageViewBean = new TrafficPageViewBean("",
                        "",
                        commonObj.getString("vc"),
                        commonObj.getString("ch"),
                        commonObj.getString("ar"),
                        commonObj.getString("is_new"),
                        0L,
                        sv,
                        1L,
                        pageObj.getLong("during_time"),
                        0L,
                        ts
                );

                return trafficPageViewBean;
            }
        });


        //2.2 uv流的转换
        SingleOutputStreamOperator<TrafficPageViewBean> uvTrafficStream = uvKafkaStream.map(new MapFunction<String, TrafficPageViewBean>() {
            @Override
            public TrafficPageViewBean map(String jsonStr) throws Exception {
                JSONObject jsonObject = JSON.parseObject(jsonStr);
                JSONObject commonObj = jsonObject.getJSONObject("common");
                JSONObject pageObj = jsonObject.getJSONObject("page");
                Long ts = jsonObject.getLong("ts");

                TrafficPageViewBean trafficPageViewBean = new TrafficPageViewBean("",
                        "",
                        commonObj.getString("vc"),
                        commonObj.getString("ch"),
                        commonObj.getString("ar"),
                        commonObj.getString("is_new"),
                        1L,
                        0L,
                        0L,
                        0L,
                        0L,
                        ts
                );

                return trafficPageViewBean;
            }
        });

        //2.2 uj流的转换
        SingleOutputStreamOperator<TrafficPageViewBean> ujTrafficStream = ujKafkaStream.map(new MapFunction<String, TrafficPageViewBean>() {
            @Override
            public TrafficPageViewBean map(String jsonStr) throws Exception {
                JSONObject jsonObject = JSON.parseObject(jsonStr);
                JSONObject commonObj = jsonObject.getJSONObject("common");
                JSONObject pageObj = jsonObject.getJSONObject("page");
              //  Long ts = jsonObject.getLong("ts");
               Long ts = jsonObject.getLong("jump_ts"); //利用跳出时间 ，来作为事件时间，进入对应的窗口

                TrafficPageViewBean trafficPageViewBean = new TrafficPageViewBean("",
                        "",
                        commonObj.getString("vc"),
                        commonObj.getString("ch"),
                        commonObj.getString("ar"),
                        commonObj.getString("is_new"),
                        0L,
                        0L,
                        0L,
                        0L,
                        1L,
                        ts
                );

                return trafficPageViewBean;
            }
        });
        //
        //3 通过union把三条流合成一个宽表流

        DataStream<TrafficPageViewBean> unionTrafficStream = pageTrafficStream.union(ujTrafficStream, uvTrafficStream);
        //unionTrafficStream.print();


        // 4  宽表的聚合
        //4.1 提取水位线
        SingleOutputStreamOperator<TrafficPageViewBean> unionTrafficWmStream = unionTrafficStream.assignTimestampsAndWatermarks(
                WatermarkStrategy.<TrafficPageViewBean>forBoundedOutOfOrderness(Duration.ofMillis(1000)).withIdleness(Duration.ofMillis(10000)).withTimestampAssigner(
                        new SerializableTimestampAssigner<TrafficPageViewBean>() {
                            @Override
                            public long extractTimestamp(TrafficPageViewBean element, long recordTimestamp) {
                                return element.getTs();
                            }
                        }
                ));

      //  unionTrafficWmStream.print();
        // 4.2 确定维度

        KeyedStream<TrafficPageViewBean, Tuple4<String, String, String, String>> tuple4KeyedStream = unionTrafficWmStream.keyBy(new KeySelector<TrafficPageViewBean, Tuple4<String, String, String, String>>() {
            @Override
            public Tuple4<String, String, String, String> getKey(TrafficPageViewBean trafficBean) throws Exception {
                return new Tuple4<String, String, String, String>(
                        trafficBean.getVc(),
                        trafficBean.getCh(),
                        trafficBean.getAr(),
                        trafficBean.getIsNew()

                );
            }
        });

        SingleOutputStreamOperator<TrafficPageViewBean> reduceStream = tuple4KeyedStream.window(TumblingEventTimeWindows.of(Time.seconds(10))).allowedLateness(Time.seconds(5))
                .reduce(new ReduceFunction<TrafficPageViewBean>() {
                            @Override
                            public TrafficPageViewBean reduce(TrafficPageViewBean bean1, TrafficPageViewBean bean2) throws Exception {
                                //把 两个bean的值相加
                                bean1.setUvCt(bean1.getUvCt() + bean2.getUvCt());
                                bean1.setUjCt(bean1.getUjCt() + bean2.getUjCt());
                                bean1.setPvCt(bean1.getPvCt() + bean2.getPvCt());
                                bean1.setDurSum(bean1.getDurSum() + bean2.getDurSum());
                                bean1.setSvCt(bean1.getSvCt() + bean2.getSvCt());
                                return bean1;
                            }
                        },
//                        new WindowFunction<TrafficPageViewBean, TrafficPageViewBean, Tuple4<String, String, String, String>, TimeWindow>() {
//                    // 要对聚合后的结果进行二次加工
//                    //  补充3个值：stt edt  ts(换成窗口的产生时间)
//                    @Override
//                    public void apply(Tuple4<String, String, String, String> tuple4, TimeWindow window, Iterable<TrafficPageViewBean> beanItr, Collector<TrafficPageViewBean> out) throws Exception {
//                        for (TrafficPageViewBean trafficPageViewBean : beanItr) {
//                            long start = window.getStart();
//                            long end = window.getEnd();
//                            String stt = DateFormatUtil.toYmdHms(start);
//                            String edt = DateFormatUtil.toYmdHms(end);
//                            trafficPageViewBean.setStt(stt);
//                            trafficPageViewBean.setEdt(edt);
//                            trafficPageViewBean.setTs(new Date().getTime());
//                            out.collect(trafficPageViewBean);
//                        }
//                    },
                        new ProcessWindowFunction<TrafficPageViewBean, TrafficPageViewBean, Tuple4<String, String, String, String>, TimeWindow>() {
                            @Override
                            public void process(Tuple4<String, String, String, String> stringStringStringStringTuple4, ProcessWindowFunction<TrafficPageViewBean, TrafficPageViewBean, Tuple4<String, String, String, String>, TimeWindow>.Context context, Iterable<TrafficPageViewBean> elements, Collector<TrafficPageViewBean> out) throws Exception {
                                for (TrafficPageViewBean trafficPageViewBean : elements) {
                                    long start = context.window().getStart();
                                    long end = context.window().getEnd();
                                    String stt = DateFormatUtil.toYmdHms(start);
                                    String edt = DateFormatUtil.toYmdHms(end);
                                    trafficPageViewBean.setStt(stt);
                                    trafficPageViewBean.setEdt(edt);
                                    trafficPageViewBean.setTs(new Date().getTime());
                                    out.collect(trafficPageViewBean);

                                }
                            }
                        });

        reduceStream.print();

//        reduceStream.addSink(ClickhouseUtil.getSink(
//                "insert into dws_traffic_vc_ch_ar_is_new_page_view_window values(?,?,?,?,?,?,?,?,?,?,?,?)"
//        ));

        env.execute();

    }
}
